package com.hzh.SparkStreaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo1WordCount {
  def main(args: Array[String]): Unit = {

    /**
     * 1、创建SparkStreaming的环境，需要依赖spark core的环境
     *
     */
    val conf = new SparkConf()
    conf.setAppName("WordCount")
    conf.setMaster("local[2]")//代表两个core，才能跑

    val sc = new SparkContext(conf)

    //创建SparkStreaming的环境,指定一个间隔时间

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    /**
     * 1、读取实时的数据源
     *
     * 安装nc
     *
     * 启动socket服务
     *
     * nc -lk 8888
     *
     */

      //得到一个DS
    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)

    val wordsDS: DStream[String] = linesDS.flatMap(_.split(","))

    val kvDS: DStream[(String, Int)] = wordsDS.map((_, 1))

    val countDS: DStream[(String, Int)] = kvDS.reduceByKey(_ + _)

    /**
     *
     * 打印结果
     */
    countDS.print()


    /**
     *
     * 启动spark Streaming程序
     */

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }
}
